package KafkaSink

import java.util
import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}

/**
 *
 * @author Spring_Hu
 * @date 2021/10/11 15:07
 */
object KaConsumer {
  def main(args: Array[String]): Unit = {
    val properties = new Properties()
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092")
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest")
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"test")
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"" +
      "org.apache.kafka.common.serialization.BytesDeserializer")
    val value = new KafkaConsumer[String, WaterSensor](properties)
    value.subscribe(util.Arrays.asList("ff"))

    while(true){
      val value1 = value.poll(1000)
     if (!value1.isEmpty) {
       value1.records("ff").forEach(record=>{
         val waterSensor = record.value()
         println(s"id =${waterSensor.id} ts=${waterSensor.ts} vc=${waterSensor.vc}")
       })
     }
    }
  }
}
